package com.fudian.spark_platform.Clustering

import com.fudian.spark_platform.Utils.AssertUtils
import org.apache.spark.mllib.clustering.LDA
import org.apache.spark.mllib.feature.HashingTF
import com.fudian.spark_platform.MLClusteringConf
import AssertUtils._
import org.apache.spark.sql.DataFrame

class LDAClustering(conf:MLClusteringConf) {


    def clustering():DataFrame = {

        val mLUtils = conf.mLUtils
        val spark = conf.spark
        val df = conf.dataF
        val tranform = conf.transformConf
        val mill = conf.millConf
        val mlConf = conf.mLConfig

        mLUtils.loadDict("new.txt")
        //通过对document的Rdd操作,取出每个文本,先进行分词操作,然后将得到的词组list和并,用空格链接各个元素后交给tokenize方法去实现过滤和特征转换
        val documents = df.select("html").rdd.map(data => {jieba(data.get(0).toString)}).map(mLUtils.tokenize)
        //获取title的值
        val titles = df.select("title").rdd.map(_.get(0).toString)
        //实例化hashingTF对象,这里调用的是spark.mllib库的hashingTF
        val hashingTF = new HashingTF()
        val mapWords = documents.flatMap(x => x)
            .map(w => (hashingTF.indexOf(w), w))
            .collect
            .toMap
        //使用documents来进行hashingTF转换
        val tf = hashingTF.transform(documents)
        //将生成的单词语index对应关系的Map传播到各个worker上去,使得改值能在各个节点上被使用
        val bcWords = tf.context.broadcast(mapWords)
        tf.cache()

//        //实例化IDF对象,调用fit方法训练IDF模型,并指出mindocs的值为2,表示至少在2个文当中出现过的词才会被计算
//        val idf = new IDF(2).fit(tf)
//        //使用hashingTF的内容来得到idf转换
//        val tf_idf: RDD[Vector] = idf.transform(tf)
//        //遍历生成的idf结果,得到每篇文章的前20个模糊主题的内容
//        val r = tf_idf.map {
//            case SparseVector(size, indices, values) =>
//                val words = indices.map(index => bcWords.value.getOrElse(index, "null"))
//                words.zip(values).sortBy(-_._2).take(20).toSeq
//        }
//        //将title和生成的结果组合,将结果写入到了本地文件系统中
//        titles.zip(r).saveAsTextFile("/Users/xiaojun/Desktop/scalaLog/mongo_zhongwen_" + System.currentTimeMillis)
//        //根据LDA的输入的需要,调用rdd的zipWithIndex方法将rdd标记后换位转换

        val rRdd = tf.zipWithIndex().map(data => {
            (data._2, data._1)
        })
        //使用LDA方法实现对idf输出的结果进行聚类
        val ldaModel = new LDA().setK(10). //设置主题个数
            setDocConcentration(-1). //设置文本
            setTopicConcentration(-1). //设置主题
            setMaxIterations(50). //设置最大迭代次数,一般情况下迭代次数越多聚类越准确
            setSeed(5L). //设置随机种子
            setCheckpointInterval(10). //设置检查点
            setOptimizer("em"). //设置检验模式
            run(rRdd)

        //获取LDA聚类后的结果
        val desTopics = ldaModel.describeTopics(5)

        //循环打印结果
        desTopics.foreach(data => {
            print("Topic  :\n")
            for (i <- Range(0, 5)) {
                val tempIndex = data._1.toList(i)
                val tempWeight = data._2.toList(i)
                println(bcWords.value.getOrElse(tempIndex, "null") + "   " + tempWeight)
                //保存至mysql
            }
        })
        //退出sparkContext,结束操作
        df
    }
}